package cn.me.alphamq.common.remote;

import cn.me.alphamq.common.entity.BaseMsg;
import cn.me.alphamq.common.entity.resp.StartSyncResp;
import cn.me.alphamq.common.store.SyncFutureManager;
import cn.me.alphamq.common.msg.TcpMsg;
import com.alibaba.fastjson2.JSON;

import java.util.concurrent.*;

/**
 * 将异步操作转化为同步操作的 future 类
 *
 * @author f
 */
public class SyncFuture implements Future<TcpMsg> {

    /**
     * 远程 rpc 返回的 tcp 消息
     */
    private TcpMsg response;

    /**
     * 消息的唯一 id
     */
    private final String msgId;

    /**
     * 让线程阻塞的工具，满足一定结果后，可以使线程恢复 TODO 可以使用 StopWatch 来代替 CountDownLatch
     */
    private final CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * 将 future 对象从管理者中移除，并且设置响应
     *
     * @param clazz 响应对象的类型，继承自 {@code BaseMsg} 类
     * @param resp  响应内容，包含响应体
     */
    public static void removeAndSetResp(Class<? extends BaseMsg> clazz, TcpMsg resp) {
        removeAndSetResp(resp.getBody(), clazz, resp);
    }

    /**
     * 将 future 对象从管理者中移除，并且设置响应
     *
     * @param body  响应体
     * @param clazz 响应对象的类型，继承自 {@code BaseMsg} 类
     * @param resp  响应内容
     */
    public static void removeAndSetResp(byte[] body, Class<? extends BaseMsg> clazz, TcpMsg resp) {
        String msgId = JSON.parseObject(body, clazz).getMsgId();
        SyncFuture syncFuture = SyncFutureManager.getAndRemove(msgId);
        if (syncFuture != null) {
            syncFuture.setResponse(resp);
        }
    }

    public SyncFuture(String msgId) {
        this.msgId = msgId;
    }

    public String getMsgId() {
        return msgId;
    }

    public TcpMsg getResponse() {
        return response;
    }

    // 设置响应结果时恢复线程运行
    public void setResponse(TcpMsg response) {
        this.response = response;
        countDownLatch.countDown();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        return false;
    }

    @Override
    public boolean isCancelled() {
        return false;
    }

    @Override
    public boolean isDone() {
        return response != null;
    }

    @Override
    public TcpMsg get() throws InterruptedException, ExecutionException {
        countDownLatch.await();
        return response;
    }

    @Override
    public TcpMsg get(long timeout, TimeUnit unit) {
        try {
            countDownLatch.await(timeout, unit);
        } catch (Exception e) {
            throw new RuntimeException(e);
        } finally {
            // 如果等待超时，需要删除指定的同步对象
            SyncFutureManager.remove(msgId);
        }
        return response;
    }
}
